package com.meiya.flink;

import com.meiya.flink.config.AppConfig;
import com.meiya.flink.service.TagProcessorService;
import com.meiya.flink.util.ConnectionTestUtil;
import com.meiya.flink.util.KafkaTestUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Flink交易处理主类
 * <p>
 * 整体数据处理流程：
 * 1. 从Kafka实时消费交易数据
 * 2. 将数据写入Doris表
 * 3. 基于clue_id和jykh的会话窗口，当窗口内没有新数据到达时触发计算
 * 4. 执行标签计算逻辑
 * 5. 将标签结果写入Kafka
 */
public class MainClass {
    private static final Logger logger = LoggerFactory.getLogger(MainClass.class);

    public static void main(String[] args) {
        logger.info("应用启动中...");

        // 解析命令行参数
        parseArgs(args);

        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        logger.info("开始启动Flink SQL标签处理任务...");
        logger.info("当前运行环境: {}", AppConfig.isProductionEnv() ? "生产环境" : "测试环境");
        logger.info("Kafka连接地址: {}", AppConfig.getKafkaBrokers());
        logger.info("Doris MySQL连接地址: {}", AppConfig.getDorisMysqlNodes());
        logger.info("Doris HTTP连接地址: {}", AppConfig.getDorisHttpNodes());
        logger.info("会话窗口间隔时间: {} 分钟", AppConfig.getSessionWindowIntervalMinutes());

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 配置表环境
        AppConfig.configureTableEnvironment(tableEnv);

        // 创建标签处理服务
        TagProcessorService tagProcessorService = new TagProcessorService(tableEnv);

        // 启动实时监听Kafka推送数据
        logger.info("开始启动实时监听Kafka推送数据...");
        try {
            tagProcessorService.startRealTimeProcessing();
            logger.info("实时监听Kafka推送数据启动完成");
        } catch (Exception e) {
            logger.error("启动实时监听Kafka推送数据失败: {}", e.getMessage(), e);
            throw e;
        }

        logger.info("所有任务已提交处理");
    }

    /**
     * 测试各种连接
     */
    private static void testConnections() {
        // 测试 Doris 连接
        ConnectionTestUtil.testDorisConnection();
        ConnectionTestUtil.testDorisHttpConnection();

        // 测试 Kafka 连接
        ConnectionTestUtil.testKafkaConnection();
        KafkaTestUtil.checkKafkaTopicData();
        KafkaTestUtil.directReadFromKafka();

        // 发送测试数据到Kafka
        KafkaTestUtil.sendTestDataToKafka();
    }

    /**
     * 解析命令行参数
     *
     * @param args 命令行参数
     */
    private static void parseArgs(String[] args) {
        if (args != null && args.length > 0) {
            for (int i = 0; i < args.length; i++) {
                // 解析环境设置参数
                if (args[i].equals("--env") && i + 1 < args.length) {
                    String env = args[i + 1].toLowerCase();
                    if (env.equals("prod") || env.equals("production")) {
                        AppConfig.setEnvironment(true);
                        logger.info("从命令行参数设置运行环境: 生产环境");
                    } else if (env.equals("test") || env.equals("dev") || env.equals("development")) {
                        AppConfig.setEnvironment(false);
                        logger.info("从命令行参数设置运行环境: 测试环境");
                    } else {
                        logger.warn("无效的环境参数: {}, 使用默认环境: {}", 
                                env, AppConfig.isProductionEnv() ? "生产环境" : "测试环境");
                    }
                    i++; // 跳过参数值
                }
                
                // 解析会话窗口间隔时间参数
                if (args[i].equals("--session-window-interval") && i + 1 < args.length) {
                    try {
                        int minutes = Integer.parseInt(args[i + 1]);
                        if (minutes > 0) {
                            AppConfig.setSessionWindowIntervalMinutes(minutes);
                            logger.info("从命令行参数设置会话窗口间隔时间: {} 分钟", minutes);
                        } else {
                            logger.warn("会话窗口间隔时间必须大于0，使用默认值: {} 分钟",
                                    AppConfig.getSessionWindowIntervalMinutes());
                        }
                    } catch (NumberFormatException e) {
                        logger.warn("无效的会话窗口间隔时间参数: {}, 使用默认值: {} 分钟",
                                args[i + 1], AppConfig.getSessionWindowIntervalMinutes());
                    }
                    i++; // 跳过参数值
                }
            }
        }
    }

}
